package flink.socket_connect;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class SocketWindowWordCount {
    public static void main(String[] args) throws Exception{
        //Flink 程序的第一步是获取运行环境 StreamExecutionEnvironment，可以用来设置参数和创建数据源以及提交任务
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //创建一个数据源
         DataStream<String> text = env.socketTextStream("192.168.0.250", 9000,"\n");

        //DataStream 是 Flink 中做流处理的核心 API，上面定义了非常多常见的操作（如，过滤、转换、聚合、窗口、关联等
        //在本示例中，我们感兴趣的是每个单词在特定时间窗口中出现的次数，比如说5秒窗口。
        // 为此，我们首先要将字符串数据解析成单词和次数（使用Tuple2表示），第一个字段是单词，第二个字段是次数，次数初始值都设置成了1。
        // 我们实现了一个 flatmap 来做解析的工作，因为一行数据中可能有多个单词。
         DataStream<WordWithCount> windowCounts = text.flatMap(new FlatMapFunction<String, WordWithCount>() {
            public void flatMap(String value, Collector<WordWithCount> out) throws Exception {
                for(String word:value.split("//s")) {
                    out.collect(new WordWithCount(word, 1));
                }
            }
         })
         .keyBy("word")
         .timeWindow(Time.seconds(5),Time.seconds(1))
         .reduce(new ReduceFunction<WordWithCount>() {
                    public WordWithCount reduce(WordWithCount value1, WordWithCount value2) throws Exception {
                        return new WordWithCount(value1.word,value1.count+value2.count);
                    }
         });
         // 将结果打印到控制台，注意这里使用的是单线程打印，而非多线程
         windowCounts.print().setParallelism(1);

        env.execute("Socket Window WordCount");
    }


    /**
     * 运行程序
     要运行示例程序，首先我们在终端(此处为"192.168.0.250 虚拟机)启动 netcat ：

     nc -lk 9000
     如果是 Windows 平台，可以通过 https://nmap.org/ncat/ 安装 ncat 然后运行：

     ncat -lk 9000

     然后直接运行SocketWindowWordCount的 main 方法。
     只需要在 netcat 控制台输入单词，就能在 SocketWindowWordCount 的输出控制台看到每个单词的词频统计。如果想看到大于1的计数，请在5秒内反复键入相同的单词。
     *
     */
}

